/**
 * Project Name:rx-java
 * File Name:AspectDemo.java
 * Package Name:top.xiaocao.rx.java.filter.function
 * Date:2024年2月13日下午3:40:01
 * Copyright (c) 2024, zhengchong.wan@minicore.com.hk All Rights Reserved.
 *
*/

package top.xiaocao.rx.java.filter.function;

import io.reactivex.rxjava3.core.Notification;
import io.reactivex.rxjava3.core.Observable;
import io.reactivex.rxjava3.core.ObservableEmitter;
import io.reactivex.rxjava3.core.ObservableOnSubscribe;
import io.reactivex.rxjava3.core.Observer;
import io.reactivex.rxjava3.disposables.Disposable;
import io.reactivex.rxjava3.functions.Action;
import io.reactivex.rxjava3.functions.Consumer;

/**
 * @className:AspectDemo 
 * @comment: 注释信息 
 * @date:     2024年2月13日 下午3:40:01 
 * @author   zhengchong.wan
 * @since    V1.0.0	 
 */
public class AspectDemo {
	
	public static void main(String[] args) {
		Observable.just(1, 2, 3)
				//执行next()事件前执行
				.doOnNext(new Consumer<Integer>() {
					@Override
					public void accept(Integer integer) throws Exception {
						System.out.println("doOnNext 事件前执行== " + integer);
					}
				})
				//当Observable每次发送一个数据事件都会调用(onNext()前回调)
				.doOnEach(new Consumer<Notification<Integer>>() {
					@Override
					public void accept(Notification<Integer> integerNotification) {
						System.out.println("doOnEach 每次都执行== " + integerNotification.getValue());
					}
				})
				//执行next()事件后执行
				.doAfterNext(new Consumer<Integer>() {
					@Override
					public void accept(Integer integer) {
						System.out.println("doAfterNext 事件后执行== " + integer);
					}
				}).subscribe(new Observer<Integer>() {
					@Override
					public void onSubscribe(Disposable d) {
						System.out.println("doNext：onSubscribe == 订阅");
					}

					@Override
					public void onNext(Integer integer) {
						System.out.println("doNext：onNext == " + integer);
					}

					@Override
					public void onError(Throwable e) {
						System.out.println("doNext：onError == " + e.getMessage());
					}

					@Override
					public void onComplete() {
						System.out.println("doNext：onComplete == ");
					}
				});
		
		System.out.println("--------------------------");
		
		Observable.create(new ObservableOnSubscribe<Integer>() {
			@Override
			public void subscribe(ObservableEmitter<Integer> e) {
				e.onNext(1);
				e.onError(new Exception("抛出异常"));
				e.onNext(2);
				e.onComplete();
			}
		})
				//Observable发送错误事件时调用
				.doOnError(new Consumer<Throwable>() {
					@Override
					public void accept(Throwable throwable) throws Exception {
						System.out.println("doOnError 发送错误事件== " + throwable.getMessage());
					}
				})
				//Observable发送事件完毕后，无论是正常发送还是异常终止都会执行
				.doAfterTerminate(new Action() {
					@Override
					public void run() throws Exception {
						System.out.println("doAfterTerminate == 事件结束");
					}
				}).subscribe(new Observer<Integer>() {
					@Override
					public void onSubscribe(Disposable d) {
						System.out.println("doOnError：onSubscribe == 订阅");
					}

					@Override
					public void onNext(Integer integer) {
						System.out.println("doOnError：onNext == " + integer);
					}

					@Override
					public void onError(Throwable e) {
						System.out.println("doOnError：onError == " + e.getMessage());
					}

					@Override
					public void onComplete() {
						System.out.println("doOnError：onComplete == ");
					}
				});
	}

}

